package rddSummary.transition.key_value_type

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object test_combineByKey {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sparkContext = new SparkContext(conf)

    /**
     * 最通用的对 key-value 型 rdd 进行聚集操作的聚集函数（aggregation function）。类似于
     * aggregate()，combineByKey()允许用户返回值的类型与输入不一致。
     * eg :
     *      将数据 List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))
     *      求每个 key 的平均值
     */

    val list: List[(String, Int)] = List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))
    val input: RDD[(String, Int)] = sparkContext.makeRDD(list, 2)
    val combineRdd: RDD[(String, (Int, Int))] = input.combineByKey(
      (_, 1),
      (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    )

    combineRdd.collect().foreach(println)

    sparkContext.stop()
  }

}
